package com.xwj.nio.reactor;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author xwj
 * @date 2020/6/11
 */
public class MultiThreadEchoHandler implements Runnable {
    private SocketChannel socketChannel;
    private SelectionKey selectionKey;
    private final ByteBuffer buf = ByteBuffer.allocate(2048);
    private static final int RECEIVING = 0, SENDING = 1;
    private int state = RECEIVING;
    private static ExecutorService executorService = Executors.newFixedThreadPool(5);

    public MultiThreadEchoHandler(Selector selector, SocketChannel sc) throws IOException {
        socketChannel = sc;
        socketChannel.configureBlocking(false);
        selectionKey = socketChannel.register(selector, 0);
        // 将handler自身作为选择键的附件
        selectionKey.attach(this);
        // 注册read就绪事件
        selectionKey.interestOps(SelectionKey.OP_READ);
        selector.wakeup();
    }

    @Override
    public void run() {
        executorService.execute(new AsyncEchoTask());
    }

    /**
     * 异步任务，不在Reactor线程中执行
     */
    public synchronized void asyncRun() {
        try {
            if (state == SENDING) {
                // 写入通道
                socketChannel.write(buf);
                // 写完后，准备开始从通道读，buf切换成写模式
                buf.clear();
                // 写完后，注册read就绪事件
                selectionKey.interestOps(SelectionKey.OP_READ);
                // 写完后，进入接收状态
                state = RECEIVING;
            } else if (state == RECEIVING) {
                // 从通道读
                int len = 0;
                while ((len = socketChannel.read(buf)) > 0) {
                    System.out.println(new String(buf.array(), 0, len));
                }
                // 读完后，准备开始写入通道
                buf.flip();
                // 读完后，注册write就绪事件
                selectionKey.interestOps(SelectionKey.OP_WRITE);
                // 读完后，进入发送的状态
                state = SENDING;
            }
            // 处理结束了，这里不能关闭select key，需要重复使用
            // sk.cancel();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 异步任务
     */
    class AsyncEchoTask implements Runnable {
        @Override
        public void run() {
            MultiThreadEchoHandler.this.asyncRun();
        }
    }
}
